package cn.azzhu.day02.transformations;

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author azzhu
 * @create 2020-09-18 01:00:37
 */
public class FoldDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //直接输入的就是单词
        final DataStreamSource<String> words = env.socketTextStream("hadoop105", 9999);

        final SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(word -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING,Types.INT));

        //在java中，认为元组是一个特殊的集合，脚标从0开始
        final KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndOne.keyBy(0);

        final SingleOutputStreamOperator<String> result = keyed.fold("", new FoldFunction<Tuple2<String, Integer>, String>() {
            @Override
            public String fold(String accumelator, Tuple2<String, Integer> value) throws Exception {
                String word = "";
                int sum = 0;
                if (accumelator.equals("")) {
                    word = value.f0;
                    sum += value.f1;
                } else {
                    final String[] fields = accumelator.split("-");
                    word = fields[0];
                    sum = Integer.parseInt(fields[1]) + value.f1;
                }
                return word + "-" + sum;
            }
        });

        result.print();

        env.execute("KeyByDemo1");
    }
}
